package com.chukun.flink.table.type;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author chukun
 * @version 1.0.0
 * @description 使用数据表字段名称
 * @createTime 2022年05月29日 22:11:00
 */
public class RowRegisterWithNames {

    public static void main(String[] args) throws Exception {

        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置批处理模式
        env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        // 设置流处理模式
        // env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

        // 创建table的环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // create a DataStream
        DataStream<Row> dataStream = env.fromElements(
                Row.of("Alice", 12),
                Row.of("Bob", 10),
                Row.of("Alice", 100));

        // 定义table字段名称
        Table namedTable = tableEnv.fromDataStream(dataStream).as("name", "score");

        tableEnv.createTemporaryView("named_table", namedTable);

        // 执行sql，返回一张结果表
        Table resultTable = tableEnv.sqlQuery(
                "SELECT name, SUM(score) FROM named_table GROUP BY name");

        // 结果表转为数据流
        DataStream<Row> resultDataStream = tableEnv.toChangelogStream(resultTable);

        resultDataStream.print("toChangelogStream");

        env.execute();
    }
}
